package com.gengzp.mq.server.job;

import com.fasterxml.jackson.core.type.TypeReference;
import com.gengzp.mq.common.constants.GengzpMqCacheConstants;
import com.gengzp.mq.common.enums.MessageStatusEnum;
import com.gengzp.mq.common.model.entity.GengzpMqMessage;
import com.gengzp.mq.common.utils.GengzpMqDateUtils;
import com.gengzp.mq.common.utils.GengzpMqJsonUtils;
import com.gengzp.mq.common.utils.GengzpMqRedisUtils;
import com.gengzp.mq.server.facade.IGengzpMqFacadeService;
import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

import static com.gengzp.mq.common.constants.GengzpMqCacheConstants.MESSAGE_KEY_PREFIX;
import static com.gengzp.mq.common.constants.GengzpMqDateFormatConstants.YMD_HMS;

/**
 * @ClassName ConsumeFailedJob
 * @Description 消息消费失败定时任务
 * @Author gengzp
 * @Date 2025/6/8 20:03
 */
@Component
public class ConsumeFailedJob {

    private static final Logger logger = LoggerFactory.getLogger(ConsumeFailedJob.class);

    @Autowired
    private IGengzpMqFacadeService gengzpMqFacadeService;

    /**
     * 消费失败扫描任务
     * 定时将处理时长超过30秒的任务进行重新入队
     * 每5分钟扫描一次
     */
    @Scheduled(fixedRate = 5 * 60 * 1000)
    public void execute() {
        // 当前系统时间
        Date startDate = new Date();
        logger.info("===================== 定时任务 [扫描处理超时消息] 开始, 当前时间: [{}] =====================",
                GengzpMqDateUtils.Date2String(startDate, YMD_HMS));
        doExecute(startDate);
        logger.info("===================== 定时任务 [扫描处理超时消息] 完成, 本次耗时: [{}ms] =====================",
                new Date().getTime() - startDate.getTime());
    }

    /**
     * 任务处理逻辑
     *
     * @param startDate 开始时间
     */
    private void doExecute(Date startDate) {
        // 从redis中取出处理中的消息id
        Object processingMessageIdsObj = GengzpMqRedisUtils.get(GengzpMqCacheConstants.PROCESSING_MESSAGE_KEY_PREFIX);
        if (Objects.isNull(processingMessageIdsObj)) {
            return;
        }
        List<Long> processingMessageIds = GengzpMqJsonUtils.parseObjectWithGeneric((String) processingMessageIdsObj, new TypeReference<>() {
        });
        if(CollectionUtils.isEmpty(processingMessageIds)){
            return;
        }

        // 根据id查询所有需要重试的消息
        List<GengzpMqMessage> messages = gengzpMqFacadeService.searchNeedRetryMessage(processingMessageIds);
        if (CollectionUtils.isEmpty(messages)) {
            return;
        }

        // 过滤出处理时间超过30秒的消息
        List<GengzpMqMessage> overTimeMessages = messages.stream()
                .filter(v -> checkIsOvertime(v.getCreatedTime(), startDate))
                .toList();
        if (CollectionUtils.isEmpty(overTimeMessages)) {
            return;
        }

        // 循环处理超时的消息
        processOvertimeMessage(overTimeMessages);
    }

    /**
     * 处理超时消息
     *
     * @param overTimeMessages 超时消息集合
     */
    private void processOvertimeMessage(List<GengzpMqMessage> overTimeMessages) {
        // 将所有消息状态修改为消费失败
        gengzpMqFacadeService.batchEditMessageStatus(
                overTimeMessages.stream().map(GengzpMqMessage::getId).collect(Collectors.toList()),
                MessageStatusEnum.CONSUMED_FAIL.getCode());

        // 重新入队的消息id
        List<Long> rePushMessageIds = Lists.newArrayList();

        for (GengzpMqMessage message : overTimeMessages) {
            logger.info("处理超时消息, 消息id: [{}]", message.getId());
            // 将消息重新入队
            GengzpMqRedisUtils.rightPush(
                    String.format("%s%s", MESSAGE_KEY_PREFIX, message.getTopic()),
                    message.getId());
            rePushMessageIds.add(message.getId());
            logger.info("将消息重新入队, 消息id: [{}]", message.getId());
        }

        // 消息重发落库
        gengzpMqFacadeService.batchRetryMessages(rePushMessageIds);
    }

    /**
     * 校验是否超时
     *
     * @param createTime  消息发送时间缓存元素对象
     * @param currentDate 当前时间
     */
    private Boolean checkIsOvertime(Date createTime, Date currentDate) {
        return currentDate.getTime() - createTime.getTime() >= 30 * 1000;
    }

}
